from fastapi import APIRouter, HTTPException, BackgroundTasks
from typing import List, Optional
from app.models.message import MessageRequest, MessageResponse, ConsumedMessage
from app.services.kafka_producer import KafkaProducerService
from app.services.kafka_consumer import KafkaConsumerService

router = APIRouter(prefix="/kafka", tags=["kafka"])

# Global instances
producer_service = KafkaProducerService()
consumer_service = KafkaConsumerService()


@router.post("/send", response_model=MessageResponse)
async def send_message(message_request: MessageRequest):
    """发送消息到 Kafka topic"""
    try:
        result = producer_service.send_message(
            topic=message_request.topic,
            message=message_request.message,
            key=message_request.key
        )
        
        if result["success"]:
            return MessageResponse(
                success=True,
                message="Message sent successfully",
                topic=result["topic"],
                partition=result["partition"],
                offset=result["offset"]
            )
        else:
            raise HTTPException(status_code=500, detail=result.get("error", "Failed to send message"))
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/consume/start")
async def start_consuming(topics: List[str], group_id: str = "fastapi-group"):
    """开始消费指定 topics 的消息"""
    try:
        result = consumer_service.start_consuming(topics, group_id)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/consume/stop")
async def stop_consuming():
    """停止消费消息"""
    try:
        result = consumer_service.stop_consuming()
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.get("/messages")
async def get_consumed_messages(limit: int = 10):
    """获取已消费的消息"""
    try:
        messages = consumer_service.get_consumed_messages(limit)
        return {"messages": messages, "count": len(messages)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.delete("/messages")
async def clear_messages():
    """清空已消费的消息"""
    try:
        result = consumer_service.clear_messages()
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.get("/topics")
async def get_topics():
    """获取所有可用的 topics"""
    try:
        topics = consumer_service.get_topics()
        return {"topics": topics}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.get("/status")
async def get_status():
    """获取 Kafka 服务状态"""
    return {
        "producer_connected": producer_service.producer is not None,
        "consumer_active": consumer_service.is_consuming,
        "consumed_messages_count": len(consumer_service.consumed_messages)
    }